package com.lili.collection.mq.plugins;

import com.lili.collection.core.context.BatchRecordContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.enums.DataFormat;
import com.lili.collection.core.error.OutputRunnerException;
import com.lili.collection.core.error.StageRecordException;
import com.lili.collection.core.plugins.OutputPlugin;
import com.lili.collection.core.record.Record;
import com.lili.collection.core.runner.result.ErrOutputData;
import com.lili.collection.core.runner.result.ErrorDetail;
import com.lili.collection.core.runner.result.RecordInfo;
import com.lili.collection.core.runner.result.RunnerResult;
import com.lili.collection.core.utils.HandlerUtil;
import com.lili.collection.core.utils.NoticeUtil;
import com.lili.collection.mq.config.PulsarmqOutputConfig;
import com.lili.collection.mq.domain.ByteMessageWrapper;
import com.lili.collection.mq.utils.JsonConvertUtil;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.*;

import java.util.List;

/**
 * 描述: pulsarmq输出插件
 *
 * @date: 2023/6/21  10:28
 * @author: lili
 */
@Slf4j
public class PulsarmqOutputPlugin implements OutputPlugin<PulsarmqOutputConfig> {
    private final static int MAX_MESSAGE_SIZE = 1024 * 1024 * 4;

    private PulsarmqOutputConfig config;
    private Throwable tempThrowable = null;
    private int executeNum;
    @Getter
    private RunnerResult runnerResult;
    @Getter
    private RunnerMsgTransfer transfer;
    private int maxMessageSize;
    private PulsarClient client;
    private Producer<String> producer;
    private boolean clientClosed;
    private boolean producerClosed;

    @Override
    public void init(PulsarmqOutputConfig config) {
        try {
            this.config = config;
            if (StringUtils.isBlank(config.getServiceUrl())) {
                throw new IllegalArgumentException("pulsarmq服务地址不能为空！");
            }
            if(this.config.getDataFormat()==null){
                throw new IllegalArgumentException("pulsarmq输入输出数据格式定义不能为空！");
            }
            ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(config.getServiceUrl());
            if (StringUtils.isNotBlank(config.getToken())) {
                clientBuilder.authentication(AuthenticationFactory.token(config.getToken()));
            }

            client = clientBuilder.build();
            if (StringUtils.isBlank(config.getTopic())) {
                throw new IllegalArgumentException("pulsarmq主题不能为空！");
            }
            this.producer = client.newProducer(Schema.STRING).topic(config.getTopic()).create();
            this.maxMessageSize = MAX_MESSAGE_SIZE;
        } catch (Exception e) {
            log.error("创建pulsarmq生产者客户端异常!", e);
            throw new OutputRunnerException(e);
        }
    }

    @Override
    public void setRunnerResult(RunnerResult outputRunnerResult) {
        this.runnerResult = outputRunnerResult;
    }

    @Override
    public void setRunnerTransfer(RunnerMsgTransfer transfer) {
        this.transfer = transfer;
    }

    @Override
    public synchronized void execute(StageContext stageContext) {
        BatchRecordContext recordContext = (BatchRecordContext) stageContext;
        RecordInfo recordInfo = new RecordInfo();
        this.executeNum++;
        RunnerResult<String> outputRunnerResult = this.getRunnerResult();
        List<Record> output = recordContext.getOutput();
        try {
            try {
                if (this.getTransfer() != null && this.getTransfer().isRunError()) {
                    throw new IllegalStateException("Rabbitmq异步运行错误!");
                }
                if (output.size() == 0) {
                    this.close(this.producer, this.client);
                    log.info("-->所有数据写入完成!");
                    return;
                }
                if (config.getDataFormat() == DataFormat.ORIGIN || config.getDataFormat()==DataFormat.ORIGIN_CONVERT) {
                    for (Record record : output) {
                        this.sendDataToPulsar(String.valueOf(record.getField().getValue()));
                        recordInfo.incrementSuccessCount();
                    }
                } else {
                    List<ByteMessageWrapper> wrapperList = JsonConvertUtil.convertRecordToJsonByteWrapper(output, recordInfo, this.maxMessageSize, false);
                    for (ByteMessageWrapper byteMessageWrapper : wrapperList) {
                        //消息推送
                        try {
                            HandlerUtil.sendProcessedData(recordInfo,config, byteMessageWrapper.getOrginJsonData(), (data) -> {
                                this.sendDataToPulsar(data);
                                return true;
                            });
                            recordInfo.addSuccessCount(byteMessageWrapper.getDataRowSize());
                            if(config.getDataFormat() != DataFormat.OBJECT){
                                recordInfo.addSuccessCount(byteMessageWrapper.getDataRowSize());
                            }
                        } catch (Throwable throwable) {
                            if(config.getDataFormat() == DataFormat.OBJECT){
                                recordInfo.addErrorCount(byteMessageWrapper.getDataRowSize()-recordInfo.getSuccessRecordCount());
                            }else {
                                recordInfo.addErrorCount(byteMessageWrapper.getDataRowSize());
                            }
                            String errorMsg = "pulsarMQ消息发送失败!";
                            recordInfo.addErrorValueDetail(new ErrorDetail("JSON", byteMessageWrapper.getOrginJsonData(), errorMsg, byteMessageWrapper.getBytes().length));
                            throw throwable;
                        }
                    }
                }
                //转换消息
                log.info("-->第{}批次数据写入完成\r\n-->写入量：{}", recordContext.getIndex(), output.size());
            } catch (Throwable throwable) {
                log.error("写入pulsarMQ错误!", throwable);
                recordInfo.getErrOutputData().putErrorInfo(throwable.toString());
                outputRunnerResult.setSuccessed(false);
                throw throwable;
            }
        } catch (Throwable throwable) {
            tempThrowable = throwable;
            RunnerMsgTransfer transfer = this.getTransfer();
            if (transfer != null) {
                transfer.setRunError(true);
                transfer.interruptRunner();
            }
            recordContext.setStageRecordState(StageContext.StageRecordState.ALL_ERROR);
            recordContext.setException(new StageRecordException(throwable));
            outputRunnerResult.setSuccessed(false);
            //错误日志收集
            ErrOutputData errOutputData = recordInfo.getErrOutputData();
            errOutputData.setBatchIndex(recordContext.getIndex());
            errOutputData.setErrorMessage(throwable.toString());
            errOutputData.setFirstBatchError(executeNum == 1 ? true : false);
            errOutputData.setType("BATCH-OUTPUT-PULSARMQ");
            log.info("-->第{}批次数据写入时发生异常\r\n-->原因：{}",
                    recordContext.getIndex(),
                    throwable.getLocalizedMessage(), throwable);
        }
        if (tempThrowable != null || !recordInfo.isSuccessed()) {
            outputRunnerResult.mergeErrorRecord(recordInfo.getErrOutputData());
        }
        this.addResultCount(recordInfo, outputRunnerResult);
        recordContext.clearOutput();
    }

    private void sendDataToPulsar(String data) throws Exception{
        if(StringUtils.isBlank(data)){
            throw new IllegalArgumentException("发送MQ数据不能为空！");
        }
        NoticeUtil.sendDataAndSaveNotice(config,data,(sendData)->{
            String primaryKey = HandlerUtil.getPrimaryKeyByData(config.getPrimaryKeySet(), sendData);
            if(StringUtils.isNotBlank(primaryKey)){
                MessageId msg = producer.newMessage().key(primaryKey).value(sendData).send();
                log.info("消息【{}】发送结果：{}",primaryKey,msg);
            }else {
                MessageId msg = producer.send(sendData);
                log.info("消息发送结果：{}",msg);
            }
            return true;
        });
    }

    private void addResultCount(RecordInfo recordInfo, RunnerResult<String> outputRunnerResult) {
        outputRunnerResult.addSuccessCount(recordInfo.getSuccessRecordCount());
        outputRunnerResult.addTotalCount(recordInfo.getTotalRecordCount());
        outputRunnerResult.addErrorCount(recordInfo.getErrorRecordCount());
        if (outputRunnerResult.isSuccessed()) {
            outputRunnerResult.setSuccessed(recordInfo.isSuccessed());
        }
    }

    @Override
    public void dispose() {
        this.close(this.producer, this.client);
    }

    /**
     * 描述: pulsar关闭客户端工具
     *
     * @date: 2023/6/21  10:46
     * @author: lili
     */
    public void close(Producer<String> producer, PulsarClient client) {
        try {
            if (producer != null && !producerClosed) {
                producer.close();
                producerClosed=true;
            }
        } catch (Exception e) {
            log.error("pulsar producer关闭异常！",e);
        }
        try {
            if (client != null && !clientClosed) {
                client.close();
                clientClosed=true;
            }
        } catch (Exception e) {
            log.error("pulsar client关闭异常！",e);
        }
    }
}